package com.hotitems_analysis

import java.sql.Timestamp
import java.util.Properties

import com.hotitems_analysis.bean.{ItemViewCount, UserBehavior}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.tuple.{Tuple, Tuple1}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Slide, Table}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/29 23:26
  */
object HotItemsWithSql {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val filePath: String = "HotItemsAnalysis/src/main/resources/UserBehavior.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    //从kafka读取
    /*val properties = new Properties()
    properties.setProperty("bootstrap.servers", "192.168.88.106:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    val kafkaStream: DataStream[String] = env
      .addSource(new FlinkKafkaConsumer[String]("hotitems", new SimpleStringSchema(), properties))
*/
    val dataStream: DataStream[UserBehavior] = fileStream.map(data => {
      val arr = data.split(",");
      UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3), arr(4).toLong)
    })
      .assignAscendingTimestamps(_.timestamp * 1000L)
    //定义表执行环境
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    //基于DataStream创建Table
    val dataTable: Table = tableEnv.fromDataStream(dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)


    //1.Table API 进行开窗聚合统计
    val aggTable: Table = dataTable
      .filter('behavior === "pv")
      .window(Slide over 1.hours every 5.minutes on 'ts as 'sw)
      .groupBy('itemId, 'sw)
      .select('itemId, 'sw.end as 'windowEnd, 'itemId.count as 'cnt)

    //用SQL实现TopN的选取
    tableEnv.createTemporaryView("aggtable", aggTable, 'itemId, 'windowEnd, 'cnt)
    val resultTable: Table = tableEnv.sqlQuery(
      """
        |select * from
        | (select *,
        | row_number() over (partition by windowEnd order by cnt desc) as row_num
        |from aggtable)
        |where row_num <= 5
      """.stripMargin)

    resultTable.toRetractStream[Row].print("table api and sql")


    //纯SQl实现
    tableEnv.createTemporaryView("datatable",dataStream,'itemId,'behavior,'timestamp.rowtime as 'ts)

    val resultSqlTable:Table = tableEnv.sqlQuery(
      """
        |select * from
        | (select *,
        | row_number() over (partition by windowEnd order by cnt desc) as row_num
        |from (
        |select itemId,
        | hop_end(ts, interval '5' minute,interval '1' hour) as windowEnd,
        | count(itemId) as cnt
        |from datatable where behavior='pv' group by itemId,
        |hop(ts, interval '5' minute,interval '1' hour)
        |))
        |where row_num <= 5
      """.stripMargin)

    resultSqlTable.toRetractStream[Row].print("sql")

    env.execute("hot items sql job")
  }


}
